package com.yzy.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

/**
 * java连接数据库的操作步骤：
 * 1.注册数据库驱动
 * 2.创建数据库连接对象
 * 3.创建数据库操作对象
 * 4.进行增删改查
 * 5.如果是查询的话，对查询的结果进行分析
 * 6.释放资源
 */

/**
 * hbase基础需求：
 * 1.如何创建一张表
 * 2.如何删除一张表
 * 3.如何向一张中添加一条数据
 * 4.如何向一张表中添加一批数据
 * 5.如何获取一条数据
 * 6.如何获取一批数据
 * 7.如何创建预分region表
 * 8.如何指定需求获取数据
 */
public class hBaseAPI {
    private Connection con;
    private Admin admin;
    private BufferedReader br = null;
    /**
     * 需求：删除一个表
     * disable 'students'
     * drop 'students'
     */

    @Before

    public void getConnection(){

        try {
            //1.获取hbase集群的配置文件对象
            //旧版本写法：0.90.0之前写法
//        HBaseConfiguration con = new HBaseConfiguration();
            Configuration conn = HBaseConfiguration.create();
            //2.因为 hbase的数据都有一条元数据，而元数存储在一张表里面，而这张元数据也是元数据，存储在zookeeper中

            conn.set("hbase.zookeeper.quorum","master:2181,node1:2181,node2:2181");
            //前提是windows中配置了hosts映射
            //3. 获取数据库的对象
            con = ConnectionFactory.createConnection(conn);

            //4.获取数据库操作对象
//            HBaseAdmin hBaseAdmin = new HBaseAdmin(con);//当你看到一个类作为方法传递的时候，传入的应该是该类的的对象或者是该类的子类对象
                                                        //HBaseAdmin(ClusterConnection connection)
                                                        //ClusterConnection extends Connection
            //新版本写法
           admin = con.getAdmin();
            System.out.println("数据库连接对象获取成功"+con);
            System.out.println("数据库操作对象获取成功"+admin);
            System.out.println("=======================");

        }catch (Exception e){
            e.printStackTrace();
        }

    }
    /**
     * 需求： 1.创建一张表
     * create 'students','info'  必须要有表名和列簇名
     */

    @Test
    public void createOneTable(){
        try {
            //先将表名封装成一个TableName对象
            TableName st = TableName.valueOf("students2");
//        HTableDescriptor students = new HTableDescriptor(tn);//在2.0.0之后过时了，并且在3.0.0之后被完全弃用
            //新版本用
//            TableDescriptorBuilder teacher= TableDescriptorBuilder.newBuilder(tn);
            TableDescriptorBuilder students2 = TableDescriptorBuilder.newBuilder(st);
            //旧版本的列簇描述对象
//        HColumnDescriptor info = new HColumnDescriptor("info");//Deprecated
            //As of release 2.0.0, this will be removed in HBase 3.0.0 (HBASE-18433 ). Use ColumnFamilyDescriptorBuilder.of(String).
            //新版本
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");

            //将列簇与表进行关联
            //旧版本中表描述器调用addColumnFamily方法将描述器添加到表中
//         students.addColumnFamily(info);
            //新版本使用setColumnFamily()
            students2.setColumnFamily(info);
            //调用方法，创建表
            // createTable(TableDescriptor desc)
            //             TableDescriptorBuilder

            admin.createTable(students2.build());
            System.out.println(st+"表创建成功！！！");

        }catch (Exception e){
            System.out.println("表创建失败！！");
            e.printStackTrace();
        }
    }

    /**
     * 需求：2。 如何删除一张表
     * disable 'students'
     */
    @Test
    public void dropOneTable(){
        try {
            //先将表名封装成一个TableName对象
            TableName tn = TableName.valueOf("teacher");

            //判断表是否存在
            if (admin.tableExists(tn)) {
                //先禁用表
                admin.disableTable(tn);
                //再使用admin的命令删除表
                //void deleteTable(TableName tableName)
                admin.deleteTable(tn);
                System.out.println(tn+"表成功删除！！");
            }else {
                System.out.println(tn+"表不存在！！");
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    /**
     * 需求：3.如何向一张表中添加一条数据
     * put 'teacher','1001','info:name','zhangsan'
     */
    @Test
    public void putOneDateToTable(){
        //先将表名封装成一个TableName对象
        TableName tn = TableName.valueOf("teacher");
        try {
            //判断表是否存在
            if (admin.tableExists(tn)){
                //获取表对象
                Table teacher = con.getTable(tn);
                //创建put对象
                //用Java方法添加数据
//            Put put = new Put("1001".getBytes());
//            put.addColumn("info".getBytes(),"name".getBytes(),"xiaohu".getBytes());)
                //用hbase自带的工具类,将字符串转字节数组
                Put put = new Put(Bytes.toBytes("1001"));
                //行键的字节数组形式
//                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes("xiaohu"));
                //新版本写法：用add添加单元格cell.
                //单元格cell是一个接口，找到它的实现类：KeyValue
                // KeyValue(byte[] row, byte[] family, byte[] qualifier, long timestamp, byte[] value)
                put.add(new KeyValue(Bytes.toBytes("1001"),
                        Bytes.toBytes("info"),
                        Bytes.toBytes("age"),
                        Bytes.toBytes("18")));
                teacher.put(put);
                System.out.println("数据添加成功！！！");

            }
        }catch (Exception e){
            System.out.println("数据添加失败！！！");
            e.printStackTrace();
        }

    }
    /**
     * 需求：4.如何向一张表添加多条数据
     */

    @Test
    public void putManyDateToTable(){

       try {
           //将表名封装成一个TableName对象
           TableName st2 = TableName.valueOf("students2");
           //创建I/O流对象
                   //创建字符缓冲流
           br = new BufferedReader(new FileReader("date/students.txt"));
           String[] colNameArray = {"","name","age","gender","clazz"};
           if (admin.tableExists(st2)) {
               //获取表对象
               Table students2 = con.getTable(st2);
               String line = null;
               while ((line=br.readLine())!=null){
                   String[] info = line.split(",");
                   byte[] rowKey = Bytes.toBytes(info[0]);
                   Put put = new Put(rowKey);
                   //行键的字节数组形式
                   for (int i = 1; i < info.length; i++) {
                        put.addColumn(Bytes.toBytes("info"),
                                Bytes.toBytes(colNameArray[i]),
                                Bytes.toBytes(info[i]));
                        students2.put(put);
                   }
               }
               System.out.println("数据传送完毕！！");
           }else{
               System.out.println("表不存在！！");
           }
       }catch (Exception e){
           e.printStackTrace();
       }

    }

    /**
     * 释放Before创建的资源，在每个Test之后执行
     */
    @After
    public void closeSource(){
        try {
            if (admin != null) {
                admin.close();
            }
            if (con != null) {
                con.close();
            }

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (br!=null){
                try {
                    br.close();
                } catch (IOException e) {

                    e.printStackTrace();
                }
            }
        }
    }

}
